package com.atlansheng.rocketmqdemoboot.demos.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atlansheng.rocketmqdemoboot.demos.po.OneOrder;
import com.atlansheng.rocketmqdemoboot.demos.service.OrderService;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.conditions.query.LambdaQueryChainWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @ClassName: TransactionListener
 * @Description:
 * @Date 2024/03/04 15:25
 * @Author songshaoqiu
 **/
@Slf4j
@RocketMQTransactionListener(rocketMQTemplateBeanName="MyRocketMqTemplete")
public class TransactionListener implements RocketMQLocalTransactionListener {

    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();

    @Autowired
    private  OrderService orderService;

    /**
     *执行事务
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.info("arg:【{}】",arg);
        log.info("执行事务");
        byte[] bytes = (byte[]) msg.getPayload();
        String string = new String(bytes);
        JSONObject jsonObject = JSON.parseObject((string));
        log.info("jsonObject{}",jsonObject);
        if (jsonObject.get("type").equals(1)){
            log.info("简单的事务消息");
            return normal(msg);
        }else{
            log.info("订单服务");
            OneOrder oneOrder = JSON.parseObject(string, OneOrder.class);
            try {
                boolean save = orderService.save(oneOrder);
                if (save){
                    return RocketMQLocalTransactionState.COMMIT;
                }
            }catch (Exception e){
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            return RocketMQLocalTransactionState.ROLLBACK;
        }
    }

    private RocketMQLocalTransactionState normal(Message msg) {
        //事务ID
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);

        int value = transactionIndex.getAndIncrement();
        log.info("值为：{}",value);
        int status = value % 3;
        assert transId != null;
        localTrans.put(transId, status);
        LocalTransactionState commitMessage = LocalTransactionState.COMMIT_MESSAGE;
        if (status == 0) {
            log.info("success");
            //成功，提交事务
            return RocketMQLocalTransactionState.COMMIT;
        }

        if (status == 1) {
            log.info("failure");
            //失败，回滚事务
            return RocketMQLocalTransactionState.ROLLBACK;
        }

        log.info("unknown");
        //中间状态
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     *检查事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        log.info("事务回查-->{}",msg);
        byte[] bytes = (byte[]) msg.getPayload();
        String string = new String(bytes);
        JSONObject jsonObject = JSON.parseObject((string));
        log.info("jsonObject{}",jsonObject);
        if (jsonObject.get("type").equals(1)){
            return normalCheck(msg);
        } else{
            log.info("订单事务回查");
            LambdaQueryWrapper<OneOrder> queryWrapper = Wrappers.lambdaQuery();
            queryWrapper.eq(OneOrder::getOrderNo, jsonObject.get("orderNo"));
            queryWrapper.le(OneOrder::getCreateTime,new Date());
            OneOrder one = orderService.getOne(queryWrapper);
            log.info("one->{}",one);
            if (Objects.isNull(one)){
                return RocketMQLocalTransactionState.ROLLBACK;
            }
            return RocketMQLocalTransactionState.COMMIT;
        }

    }

    private RocketMQLocalTransactionState normalCheck(Message msg) {
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
        Integer status = localTrans.get(transId);
        if (null != status) {
            switch (status) {
                case 0:
                    retState = RocketMQLocalTransactionState.COMMIT;
                    break;
                case 1:
                    retState = RocketMQLocalTransactionState.ROLLBACK;
                    break;
                case 2:
                    retState = RocketMQLocalTransactionState.UNKNOWN;
                    break;

                default: break;
            }
        }
        log.info("msgTransactionId:{},TransactionState:{},status:{}",transId,retState,status);
        return retState;
    }
}